"""
Case Type   : 基础功能
Case Name   : 使用pitr恢复至创建发布订阅之前
Create At   : 2022/3/21
Owner       : opentestcase026
Description :
    1.两集群开启归档，并进行基础备份
    2.在两个集群创建表
    3.两个集群均创建laber_name
    4.创建发布订阅
    5.更新数据
    6.查询数据是否更新
    7.发布端恢复至laber_name
    8.查询发布端是否存在
    9.重新创建发布
    10.更新并查询数据是否更新
    11.订阅端恢复至laber_name
    12.查询订阅是否存在
    13.重新创建订阅
    14.插入并查询是否更新
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.成功
    6.更新
    7.成功
    8.不存在
    9.成功
    10.插入数据成功，不更新，复制槽不存在
    11.成功
    12.不存在
    13.成功
    14.插入数据成功，且更新成功
History     :
    Modified by opentestcase026 2022/5/5:增加等待5.5s，适配最新代码
    Modified by opentestcase004 2022/8/02:增加基础数据同步功能后订阅端有基础数据的同
    时再在订阅端修改表数据,主键冲突,关闭基础数据同步
    Modified by opentestcase004 2022/8/12:清理环境重启库添加断言
    Modified by opentestcase004 2022/9/23:公共函数参数限制，优化用例对订阅端备机的重
    建方式
    Modified by opentestcase012 2022/10/11:优化用例和断言,添加查看数据库状态
    Modified by opentestcase012 2022/11/9 优化用例，build之后添加等待主备同步
"""
import unittest
import os
import time
from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.com_sub = Common('remote1_PrimaryDbUser')
        self.tb_name1 = 'tb_pubsub_case073_1'
        self.tb_name2 = 'tb_pubsub_case073_2'
        self.tbs_name = 'tps_pubsub_case073'
        self.schema_name1 = 's_pubsub_case073_1'
        self.subname1 = "sub_case073_1"
        self.pubname1 = "pub_case073_1"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1)
        self.hba_pub = os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')
        self.hba_sub = os.path.join(macro.DB_INSTANCE_PATH_REMOTE1,
                                    'pg_hba.conf')
        self.hba_bak_pub = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH), 'pg_hba.conf')
        self.hba_bak_sub = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1), 'pg_hba.conf')
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'

        self.backup_path_pub = os.path.join(self.parent_path_pub,
                                            'back_pubsub_case073')
        self.backup_path_sub = os.path.join(self.parent_path_sub,
                                            'back_pubsub_case073')
        self.comsh_sta_pub = []
        self.nodelist_pub = ['Standby1DbUser', 'Standby2DbUser']
        self.num_pub = int(self.commsh_pub.get_node_num()) - 1
        self.log.info(str(self.num_pub))
        for i in range(self.num_pub):
            self.comsh_sta_pub.append(CommonSH(self.nodelist_pub[i]))
        self.comsh_sta_sub = []
        self.nodelist_sub = ['remote1_Standby1DbUser',
                             'remote1_Standby2DbUser']
        for i in range(len(self.nodelist_sub)):
            self.comsh_sta_sub.append(CommonSH(self.nodelist_sub[i]))
        self.tbs_path_pub = os.path.join(self.parent_path_pub, self.tbs_name)
        self.tbs_path_sub = os.path.join(self.parent_path_sub, self.tbs_name)
        self.archive_path_pub = os.path.join(self.parent_path_pub, 'archive')
        self.archive_path_sub = os.path.join(self.parent_path_sub, 'archive')
        self.recovery_path_pub = os.path.join(macro.DB_INSTANCE_PATH,
                                              'recovery.conf')
        self.recovery_path_sub = os.path.join(macro.DB_INSTANCE_PATH_REMOTE1,
                                              'recovery.conf')
        self.archive_mode_pub = self.com_pub.show_param("archive_mode")
        self.enable_slot_log_pub = self.com_pub.show_param("enable_slot_log")
        self.archive_mode_sub = self.com_sub.show_param(
            "archive_mode", macro.DB_ENV_PATH_REMOTE1)
        self.enable_slot_log_sub = self.com_sub.show_param(
            "enable_slot_log", macro.DB_ENV_PATH_REMOTE1)
        self.hostname_pub = self.pri_userdb_pub.sh('hostname').result()
        self.hostname_sub = self.pri_userdb_sub.sh('hostname').result()

        text = '-----预置条件1:备份配置文件pg_hba.conf expect:状态正常-----'
        self.log.info(text)
        cmd = f'cp {self.hba_pub} {self.hba_bak_pub};'
        self.log.info(cmd)
        msg1 = self.com_pub.get_sh_result(self.pri_userdb_pub, cmd)
        cmd = f'cp {self.hba_sub} {self.hba_bak_sub};'
        self.log.info(cmd)
        msg2 = self.com_pub.get_sh_result(self.pri_userdb_sub, cmd)
        self.assertEqual('', msg1 and msg2, '执行失败:' + text)

        text = '-----预置条件2:查看发布端和订阅端的数据库状态 expect:状态正常-----'
        self.log.info(text)
        status = self.commsh_pub.get_db_cluster_status()
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)
        status = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        text = '-----预置条件3:修改发布端pg_hba文件和配置参数wal_level,重启数据库 expect:成功-----'
        self.log.info(text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)
        if 'logical' != self.wal_level:
            result = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(result, '执行失败:' + text)
            result = self.commsh_pub.restart_db_cluster(False)
            self.assertTrue(result, '执行失败:' + text)
            status = self.commsh_pub.get_db_cluster_status()
            self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status,
                          '执行失败:' + text)

        text = '-----预置条件4:订阅端修改pg_hba文件 expect:成功-----'
        self.log.info(text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败:' + text)

    def test_pubsub(self):
        text = '--step1:两集群开启归档，并进行基础备份 expect:成功--'
        self.log.info(text)
        self.log.info("发布端:\n")
        cmd = f"mkdir {self.backup_path_pub};mkdir {self.archive_path_pub};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'archive_mode=on',
            self.hostname_pub)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f"archive_command = 'cp %p {self.archive_path_pub}/%f'",
            self.hostname_pub)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'enable_slot_log=on')
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_pub.restart_db_cluster(False)
        self.assertTrue(result, '执行失败:' + text)
        status = self.commsh_pub.get_db_cluster_status()
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        time.sleep(10)
        result = self.commsh_pub.exec_gs_basebackup(
            self.backup_path_pub, self.pri_userdb_pub.ssh_host,
            self.pri_userdb_pub.db_port, )
        self.assertTrue(result, '执行失败:' + text)
        cmd = f"chmod 700 {self.backup_path_pub}"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        self.log.info("订阅端：\n")
        cmd = f"mkdir {self.backup_path_sub};mkdir {self.archive_path_sub};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'archive_mode=on',
            self.hostname_sub, dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f"archive_command = 'cp %p {self.archive_path_sub}/%f'",
            self.hostname_sub, dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'enable_slot_log=on',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        result = self.commsh_sub.restart_db_cluster(
            False, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        status = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        time.sleep(10)
        result = self.commsh_sub.exec_gs_basebackup(
            self.backup_path_sub, self.pri_userdb_sub.ssh_host,
            self.pri_userdb_sub.db_port,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        cmd = f"chmod  700 {self.backup_path_sub}"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)

        text = '--step2:两个集群均创建表 expect:成功--'
        self.log.info(text)
        sql1 = f"drop tablespace if exists {self.tbs_name};" \
            f"create tablespace {self.tbs_name} " \
            f" location '{self.tbs_path_pub}';"
        sql2 = f"drop tablespace if exists {self.tbs_name};" \
            f"create tablespace {self.tbs_name} " \
            f" location '{self.tbs_path_sub}';"
        sql = f"drop schema if exists {self.schema_name1};" \
            f"create schema {self.schema_name1};" \
            f"drop table if exists {self.schema_name1}.{self.tb_name1};" \
            f"create table {self.schema_name1}.{self.tb_name1}" \
            f"(id1 int primary key, id2 int, id3 int) " \
            f"tablespace {self.tbs_name};" \
            f"drop table if exists {self.tb_name2};" \
            f"create table {self.tb_name2}(id int primary key " \
            f"constraint id_nn not null,use_filename varchar2(20)," \
            f"filename varchar2(255),text varchar2(2000))" \
            f"PARTITION BY RANGE(id)" \
            f"(partition p1 values less than(30), " \
            f"partition p2 values less than(60)," \
            f"partition p3 values less than(90)," \
            f" partition p4 values less than(maxvalue));"
        result = self.commsh_pub.execut_db_sql(
            sql1 + sql, sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         5, '执行失败:' + text)
        result = self.commsh_sub.execut_db_sql(sql2 + sql, self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.TABLE_CREATE_SUCCESS),
                         5, '执行失败:' + text)

        text = '--step3:两个集群均创建laber_name expect:成功--'
        self.log.info(text)
        sql = "select pg_switch_xlog();" \
              "select pg_create_restore_point('point1');"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn('pg_create_restore_point', result, '执行失败' + text)
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('pg_create_restore_point', result, '执行失败' + text)

        text = '--step4:创建发布订阅 expect:成功--'
        self.log.info(text)
        sql = f"drop publication if exists {self.pubname1};" \
            f"create publication {self.pubname1} for all tables;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql = f"drop subscription if exists {self.subname1};" \
            f"create subscription {self.subname1} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"publication {self.pubname1} with (copy_data=false);"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.create_sub_succ_msg), 1,
                         '执行失败:' + text)

        text = '--step5:更新数据 expect:成功--'
        self.log.info(text)
        sql_insert = f"insert into {self.schema_name1}.{self.tb_name1} " \
            f"values(generate_series(1,100), 1, 1);" \
            f"insert into {self.tb_name2} values(1, 'first', '%一', '')," \
            f"(60, 'first', '%二', ''),(90, 'first', '%三', ''); " \
            f"select pg_sleep(5.5);"
        result = self.commsh_pub.execut_db_sql(sql_insert,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.INSERT_SUCCESS_MSG),
                         2, '执行失败' + text)

        text = "--step6:查询数据是否更新 expect:更新--"
        self.log.info(text)
        sql_select = f"select count(*) from " \
            f"{self.schema_name1}.{self.tb_name1};" \
            f"select * from {self.tb_name2};"
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('3 rows', result, '执行失败' + text)
        self.assertIn('1 | first        | %一     |', result, '执行失败' +
                      text)
        result_a = self.commsh_pub.execut_db_sql(sql_select,
                                                 sql_type=self.user_param_pub)
        self.log.info(result_a)
        self.assertEqual(result_a, result, '执行失败' + text)

        text = "--step7:发布端恢复至laber_name expect:更新--"
        self.log.info(text)
        result = self.commsh_pub.stop_db_cluster()
        self.assertTrue(result, '执行失败:' + text)
        cmd = f"mv {macro.DB_INSTANCE_PATH} {macro.DB_INSTANCE_PATH}_bak;" \
            f"mv {self.backup_path_pub} {macro.DB_INSTANCE_PATH};" \
            f"ls -al {self.archive_path_pub}"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"touch {self.recovery_path_pub};" \
            f"echo \"restore_command = " \
            f"'cp {self.archive_path_pub}/%f %p'\" > " \
            f"{self.recovery_path_pub};" \
            f"echo \"recovery_target_name='point1'\" >> " \
            f"{self.recovery_path_pub};" \
            f"cat {self.recovery_path_pub};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_pub.start_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        time.sleep(20)
        cmd_resume = "select pg_is_in_recovery();" \
                     "select pg_xlog_replay_resume();select pg_sleep(10);" \
                     "select pg_is_in_recovery();"
        result = self.commsh_pub.execut_db_sql(cmd_resume)
        self.log.info(result)
        self.assertEqual(19, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('f', result.splitlines()[-2], '执行失败:' + text)
        for i in range(self.num_pub):
            result = self.comsh_sta_pub[i].build_standby("-t 3600")
            self.assertIn(self.constant.BUILD_SUCCESS_MSG,
                          result, '执行失败:' + text)

        text = '-----校验备机是否完成数据同步-----'
        self.log.info(text)
        check_msg = self.commsh_pub.check_location_consistency(
            'primary', self.commsh_pub.get_node_num())
        self.assertTrue(check_msg, '执行失败:' + text)
        status = self.commsh_pub.get_db_cluster_status()
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)
        cmd = f"ls -al  {macro.DB_INSTANCE_PATH}"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_pub.execut_db_sql(sql_select,
                                               sql_type=self.user_param_pub)
        self.log.info(result)

        text = '--step8:查询发布端是否存在 expect:不存在--'
        self.log.info(text)
        sql = "select * from pg_publication;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertNotIn(self.pubname1, result, '执行失败' + text)

        text = '--step9:重新创建发布 expect:成功--'
        self.log.info(text)
        sql = f"create publication {self.pubname1} for all tables;"
        result = self.commsh_pub.execut_db_sql(sql,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertIn(self.constant.create_pub_succ_msg, result,
                      '执行失败:' + text)
        self.assertNotIn(self.constant.SQL_WRONG_MSG[1], result,
                         '执行失败:' + text)

        text = '--step10:更新并查询数据是否更新 expect:插入数据成功，不更新，' \
               '复制槽不存在--'
        self.log.info(text)
        sql_insert = f"insert into {self.schema_name1}.{self.tb_name1} " \
            f"values(200, 1, 1);" \
            f"insert into {self.tb_name2} values" \
            f"(200, 'first', '%一', 'new');" \
            f"select pg_sleep(30);" \
            f"select active from pg_replication_slots where " \
            f"slot_name='{self.subname1}';"
        result = self.commsh_pub.execut_db_sql(sql_insert,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.INSERT_SUCCESS_MSG),
                         2, '执行失败' + text)
        self.assertEqual(10, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('0 rows', result.splitlines()[-1].strip(),
                      '执行失败' + text)
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn('3 rows', result, '执行失败' + text)
        self.assertIn('1 row', result, '执行失败' + text)
        self.assertNotIn('101', result, '执行失败' + text)
        self.assertNotIn('200 | first        | %一     | new',
                         result, '执行失败' + text)

        text = "--step11:订阅端恢复至laber_name expect:更新--"
        self.log.info(text)
        result = self.commsh_sub.stop_db_cluster(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(result, '执行失败:' + text)
        cmd = f"mv {macro.DB_INSTANCE_PATH_REMOTE1} " \
            f"{macro.DB_INSTANCE_PATH_REMOTE1}_bak;" \
            f"mv {self.backup_path_sub} {macro.DB_INSTANCE_PATH_REMOTE1};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        cmd = f"touch {self.recovery_path_sub};" \
            f"echo \"restore_command = " \
            f"'cp {self.archive_path_sub}/%f %p'\" > " \
            f"{self.recovery_path_sub};" \
            f"echo \"recovery_target_name='point1'\" >> " \
            f"{self.recovery_path_sub};" \
            f"cat {self.recovery_path_sub};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        result = self.commsh_sub.start_db_cluster(
            True, env_path=macro.DB_ENV_PATH_REMOTE1)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败:' + text)
        time.sleep(20)
        result = self.commsh_sub.execut_db_sql(cmd_resume,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(19, len(result.splitlines()), '执行失败:' + text)
        self.assertIn('f', result.splitlines()[-2], '执行失败:' + text)
        result = self.commsh_sub.get_standby_and_build(
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertIn(self.constant.BUILD_SUCCESS_MSG, str(result),
                      '执行失败:' + text)
        status = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status, '执行失败:' + text)

        text = '--step12:查询订阅是否存在 expect:不存在--'
        self.log.info(text)
        sql = "select * from pg_subscription;"
        result = self.commsh_sub.execut_db_sql(sql,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertNotIn(self.subname1, result, '执行失败:' + text)

        text = '--step13:重新创建订阅 expect:成功--'
        self.log.info(text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn('', result, '执行失败:' + text)
        sql = f"create subscription {self.subname1} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} " \
            f"user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' " \
            f"publication {self.pubname1} with (copy_data=false);"
        result = self.commsh_sub.execut_db_sql(sql, self.user_param_sub, None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.create_sub_succ_msg), 1,
                         '执行失败:' + text)

        text = '--step14:插入并查询是否更新 expect:插入数据成功，且更新成功--'
        self.log.info(text)
        sql_insert = f"insert into {self.schema_name1}.{self.tb_name1} " \
            f"values(222, 1, 1);" \
            f"insert into {self.tb_name2} values(222, 'first', '%一', 'new');" \
            f"select pg_sleep(30);" \
            f"select * from pg_replication_slots;"
        result = self.commsh_pub.execut_db_sql(sql_insert,
                                               sql_type=self.user_param_pub)
        self.log.info(result)
        self.assertEqual(result.count(self.constant.INSERT_SUCCESS_MSG),
                         2, '执行失败' + text)
        sql_select = f"select * from " \
            f"{self.schema_name1}.{self.tb_name1};" \
            f"select * from {self.tb_name2};"
        result = self.commsh_sub.execut_db_sql(sql_select,
                                               self.user_param_sub,
                                               None,
                                               macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result)
        self.assertEqual(result.count('1 row'), 2, '执行失败' + text)
        self.assertIn('222 | first        | %一     | new',
                      result, '执行失败' + text)
        self.assertIn('222 |   1 |   1', result, '执行失败' + text)

    def tearDown(self):
        self.log.info('-----清理环境-----')
        text1 = '-----发布端数据库重建备机查看集群状态 expect:成功-----'
        self.log.info(text1)
        status1_pub = self.commsh_pub.get_db_cluster_status('status')
        self.log.info(status1_pub)
        if not status1_pub:
            check_cmd = \
                f'if [[ -d {macro.DB_INSTANCE_PATH} && -d ' \
                    f'{macro.DB_INSTANCE_PATH}_bak ]]; ' \
                    f'then echo "exists"; else echo "not exists"; fi'
            self.log.info(check_cmd)
            result = self.pri_userdb_pub.sh(check_cmd).result()
            if result == 'exists':
                cmd = f"rm -rf {macro.DB_INSTANCE_PATH_REMOTE1};"
                result = self.pri_userdb_pub.sh(cmd).result()
                self.log.info(result)
            cmd = f"mv {macro.DB_INSTANCE_PATH}_bak {macro.DB_INSTANCE_PATH};"
            self.log.info(cmd)
            result = self.pri_userdb_pub.sh(cmd).result()
            self.log.info(result)
            self.commsh_pub.restart_db_cluster()
            for i in range(self.num_pub):
                self.comsh_sta_pub[i].build_standby("-t 3600")
            status1_pub = self.commsh_pub.get_db_cluster_status('status')
            self.log.info(status1_pub)

        text2 = '-----订阅端数据库重建备机查看集群状态 expect:成功-----'
        self.log.info(text2)
        status1_sub = self.commsh_sub.get_db_cluster_status(
            'status', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status1_sub)
        if not status1_sub:
            check_cmd = \
                f'if [[ -d {macro.DB_INSTANCE_PATH} && -d ' \
                    f'{macro.DB_INSTANCE_PATH}_bak ]]; ' \
                    f'then echo "exists"; else echo "not exists"; fi'
            self.log.info(check_cmd)
            result = self.pri_userdb_sub.sh(check_cmd).result()
            if result == 'exists':
                cmd = f"rm -rf {macro.DB_INSTANCE_PATH_REMOTE1};"
                result = self.pri_userdb_sub.sh(cmd).result()
                self.log.info(result)
            cmd = f"mv {macro.DB_INSTANCE_PATH_REMOTE1}_bak " \
                f"{macro.DB_INSTANCE_PATH_REMOTE1};"
            self.log.info(cmd)
            result = self.pri_userdb_sub.sh(cmd).result()
            self.log.info(result)
            self.commsh_sub.restart_db_cluster(True,
                                               macro.DB_ENV_PATH_REMOTE1)
            build_result = self.commsh_sub.get_standby_and_build(
                dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
                env_path=macro.DB_ENV_PATH_REMOTE1)
            self.log.info(build_result)
            status1_sub = self.commsh_sub.get_db_cluster_status(
                'status', env_path=macro.DB_ENV_PATH_REMOTE1)
            self.log.info(status1_sub)

        text3 = '-----删除发布订阅 expect:成功-----'
        self.log.info(text3)
        sql = f"drop subscription if exists {self.subname1};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)
        sql = f"drop publication if exists {self.pubname1};" \
            f"purge recyclebin"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)

        text4 = '-----订阅端和发布端删除表,schema和表空间 expect:成功-----'
        self.log.info(text4)
        sql = f"drop table if exists {self.schema_name1}.{self.tb_name1};" \
            f"drop table if exists {self.tb_name2};" \
            f"drop schema {self.schema_name1};" \
            f"drop tablespace {self.tbs_name};"
        sql_drop_sub = self.commsh_sub.execut_db_sql(sql, self.user_param_sub,
                                                     None,
                                                     macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sql_drop_sub)
        sql_drop_pub = self.commsh_pub.execut_db_sql(
            sql, sql_type=self.user_param_pub)
        self.log.info(sql_drop_pub)

        text5 = '-----订阅端和发布端恢复pg_hba.conf文件 expect:成功-----'
        self.log.info(text5)
        cmd = f'if [[ -e {self.hba_pub} && -e {self.hba_bak_pub} ]]; ' \
            f'then rm -rf {self.hba_pub}; ' \
            f'mv {self.hba_bak_pub} {self.hba_pub}; ' \
            f'else echo not exits; fi'
        hba_msg_pub = self.com_pub.get_sh_result(self.pri_userdb_pub, cmd)
        cmd = f'if [[ -e {self.hba_sub} && -e {self.hba_bak_sub} ]]; ' \
            f'then rm -rf {self.hba_sub}; ' \
            f'mv {self.hba_bak_sub} {self.hba_sub}; ' \
            f'else echo not exits; fi'
        hba_msg_sub = self.com_pub.get_sh_result(self.pri_userdb_sub, cmd)

        text6 = '-----订阅端和发布端删除备份文件 expect:成功-----'
        self.log.info(text6)
        cmd = f"rm -rf {self.backup_path_pub};" \
            f"rm -rf {macro.DB_INSTANCE_PATH}_bak;" \
            f"rm -rf {self.tbs_path_pub};" \
            f"rm -rf {self.archive_path_pub};" \
            f"rm -rf {self.recovery_path_pub}"
        self.log.info(cmd)
        rm_msg_pub = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(rm_msg_pub)
        cmd = f"rm -rf {self.tbs_path_sub};" \
            f"rm -rf {self.backup_path_sub};" \
            f"rm -rf {macro.DB_INSTANCE_PATH_REMOTE1}_bak;" \
            f"rm -rf {self.archive_path_sub};" \
            f"rm -rf {self.recovery_path_sub}"
        self.log.info(cmd)
        rm_msg_sub = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(rm_msg_sub)

        text7 = '-----订阅端和发布端恢复参数配置 expect:成功-----'
        self.log.info(text7)
        guc_wal_pub = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        guc_archive_pub = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'archive_mode={self.archive_mode_pub}',
            self.hostname_pub)
        guc_slot_pub = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_slot_log={self.enable_slot_log_pub}')
        guc_archive_sub = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'archive_mode={self.archive_mode_sub}',
            self.hostname_sub, dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        guc_slot_sub = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_slot_log={self.enable_slot_log_sub}',
            self.hostname_sub, dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)

        text8 = '-----重启数据库查看数据库状态 expect:成功-----'
        self.log.info(text8)
        res_msg_pub = self.commsh_pub.restart_db_cluster(False)
        status_pub = self.commsh_pub.get_db_cluster_status()
        self.log.info(status_pub)
        res_msg_sub = self.commsh_sub.restart_db_cluster(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        status_sub = self.commsh_sub.get_db_cluster_status(
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status_sub)

        self.assertTrue(status1_pub, '执行失败:' + text1)
        self.assertTrue(status1_sub, '执行失败:' + text2)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text3)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text3)
        self.assertIn('PURGE RECYCLEBIN', drop_pub_result, '执行失败' + text3)
        self.assertEqual(3, sql_drop_sub.count(
            self.constant.TABLE_DROP_SUCCESS), '执行失败' + text4)
        self.assertEqual(3, sql_drop_pub.count(
            self.constant.TABLE_DROP_SUCCESS), '执行失败' + text4)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, sql_drop_sub,
                      '执行失败' + text4)
        self.assertIn(self.constant.DROP_SCHEMA_SUCCESS_MSG, sql_drop_pub,
                      '执行失败' + text4)
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, sql_drop_sub,
                      '执行失败' + text4)
        self.assertIn(self.constant.TABLESPCE_DROP_SUCCESS, sql_drop_pub,
                      '执行失败' + text4)
        self.assertEqual('', hba_msg_pub and hba_msg_sub, '执行失败:' + text5)
        self.assertEqual('', rm_msg_pub and rm_msg_sub, '执行失败:' + text6)
        self.assertTrue(guc_wal_pub, '执行失败:' + text7)
        self.assertTrue(guc_archive_pub, '执行失败:' + text7)
        self.assertTrue(guc_slot_pub, '执行失败:' + text7)
        self.assertTrue(guc_archive_sub, '执行失败:' + text7)
        self.assertTrue(guc_slot_sub, '执行失败:' + text7)
        self.assertTrue(res_msg_pub, '执行失败:' + text8)
        self.assertTrue(res_msg_sub, '执行失败:' + text8)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status_pub,
                      '执行失败:' + text8)
        self.assertIn(self.constant.CLUSTER_NORMAL_MSG, status_sub,
                      '执行失败:' + text8)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
